package mr

import (
	"fmt"
	"log"
	"sync"
	"time"
)
import "net"
import "os"
import "net/rpc"
import "net/http"

const (
	WAITING  = 0
	IDLE     = 1
	INGRESS  = 2
	FINISHED = 3
	TTL      = 10
	MAP      = 0
	REDUCE   = 1
	KILL     = 2
	WAIT     = 3
)

var lock = new(sync.Mutex)

type Master struct {
	// Your definitions here.
	tasks     []*Task
	mapId     int
	reduceId  int
	completed bool
	nReduce   int
}

type Task struct {
	command   int
	id        int
	status    int
	nReduce   int
	startTime time.Time
	file      string
}

func (task Task) String() string {
	return fmt.Sprintf("c:%v i:%v s:%v", task.command, task.id, task.status)
}

func (task *Task) isIdle() bool {
	return task.status == IDLE
}
func (task *Task) isWaiting() bool {
	return task.status == WAITING
}
func (task *Task) isIngress() bool {
	return task.status == INGRESS
}
func (task *Task) isFinished() bool {
	return task.status == FINISHED
}
func (task *Task) isOverTime() bool {
	return task.isIngress() && task.startTime.Add(time.Second*TTL).Before(time.Now())
}

// Your code here -- RPC handlers for the worker to call.

//
// an example RPC handler.
//
// the RPC argument and reply types are defined in rpc.go.
//
func (m *Master) Example(args *ExampleArgs, reply *ExampleReply) error {
	reply.Y = args.X + 1
	return nil
}
func (m *Master) Submit(args *SubmitArgs, reply *SubmitReply) error {
	lock.Lock()
	defer lock.Unlock()
	for _, task := range m.tasks {
		if args.Command != task.command || args.Id != task.id {
			continue
		}
		if task.isIngress() {
			task.status = FINISHED
			m.updateCompleted()

			if task.command == MAP {
				m.updateReduceStatus()
			}
		}
	}
	return nil
}
func (m *Master) FetchTask(args *FetchTaskArgs, reply *FetchTaskReply) error {
	lock.Lock()
	defer lock.Unlock()
	//fmt.Printf("tasks->%+q\n", m.tasks)
	if m.completed {
		reply.Command = KILL
		return nil
	}

	for index, task := range m.tasks {
		if task.isIdle() || task.isOverTime() {
			reply.Command = task.command
			reply.Id = task.id
			reply.NReduce = task.nReduce
			reply.File = task.file

			m.tasks[index].status = INGRESS
			m.tasks[index].startTime = time.Now()
			return nil
		}
	}

	reply.Command = WAIT
	return nil
}

//
// start a thread that listens for RPCs from worker.go
//
func (m *Master) server() {
	rpc.Register(m)
	rpc.HandleHTTP()
	//l, e := net.Listen("tcp", ":1234")
	sockname := masterSock()
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

//
// main/mrmaster.go calls Done() periodically to find out
// if the entire job has finished.
//
func (m *Master) Done() bool {
	return m.completed
}

func (m *Master) updateCompleted() {
	for _, task := range m.tasks {
		if !task.isFinished() {
			m.completed = false
			return
		}
	}
	m.completed = true
}

func (m *Master) updateReduceStatus() {
	if m.allMapTaskFinished() {
		for _, task := range m.tasks {
			if task.command == REDUCE {
				task.status = IDLE
			}
		}
	}
}

func (m *Master) allMapTaskFinished() bool {
	for _, task := range m.tasks {
		if task.command == MAP && !task.isFinished() {
			return false
		}
	}
	return true
}

//
// create a Master.
// main/mrmaster.go calls this function.
// nReduce is the number of reduce tasks to use.
//
func MakeMaster(files []string, nReduce int) *Master {
	m := Master{
		tasks:   []*Task{},
		nReduce: nReduce,
	}

	// Your code here.
	for _, file := range files {
		m.tasks = append(m.tasks, &Task{
			command: MAP,
			id:      m.mapId,
			status:  IDLE,
			nReduce: m.nReduce,
			file:    file,
		})
		m.mapId += 1
	}

	for i := 0; i < nReduce; i++ {
		m.tasks = append(m.tasks, &Task{
			command: REDUCE,
			id:      m.reduceId,
			status:  WAITING,
		})
		m.reduceId += 1
	}

	m.server()
	return &m
}
